package cc.verywell.pureblock.msg.p2p;

import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import cc.verywell.pureblock.func.FuncIndex;
import cc.verywell.pureblock.func.FuncProvider;
import cc.verywell.pureblock.func.FuncTimelog;
import cc.verywell.pureblock.func.node.Node;
import cc.verywell.pureblock.func.node.NodeBucket;
import cc.verywell.pureblock.func.node.NodeBucket.ValueShell;
import cc.verywell.pureblock.func.node.NodesMsg;
import cc.verywell.pureblock.log.Pblog;
import cc.verywell.pureblock.net.Msg;
import cc.verywell.pureblock.net.NormalSession;
import cc.verywell.pureblock.net.Timelog;
import cc.verywell.pureblock.util.StringUtil;

/**
* 
* Copyright 2019 TanYaqiu
* @author TanYaqiu 
* @address flat 601
* @date 2019年3月18日 下午11:20:53
*/
public class P2pFuncProvider implements FuncProvider,NodeBucket.NewKeyListener {
	static public int PEERS_TAKE_MAX = 16;
	
	static final private byte TimePing = (byte)1;
	static final private byte TimePong = (byte)2;
	static final private byte Goodbye = (byte)3;
	static final private byte PeersWant = (byte)4;
	static final private byte PeersTake = (byte)5;

	final private NodeBucket nodeBucket;

	public P2pFuncProvider(NodeBucket nodeBucket) {
		this.nodeBucket = nodeBucket;
		this.pinging = new ConcurrentHashMap<>();
		this.nodeBucket.addNewKeyListener(this);
	}
	@Override
	public void newKey(Collection<ValueShell> keys) {
		//TODO 这里可以加过滤表，使得PeersTake只返回未发送过的Node;
	}
	
	/**
	 * 解包线程，这里出错会关掉channel;
	 * <p>此时不要用close()中断网络，因为就算中断，后续还会继续解包的，直到没数据了，或者进入一次toDo</p>
	 * <p>实在要关闭，请throw Exception</p>
	 * <p>解包或跳过均可，否则剩下的remain会进入下一轮decode</p>
	 * @param bb
	 * @return
	 */
	@Override
	public Msg decode(ByteBuffer bb, NormalSession nc) {
		//int code = 0xFF & bb.get();
		byte b = bb.get();
		//Pblog.IO("p2p:",(0xFF & b));
		switch(b) {
		case TimePing:{
			long now = System.currentTimeMillis();
			Timelog tl = nc.getTimelog();
			long time = tl.getLastReadTime();
			if(time<=0)time = now;
			TimePingMsg ping = new TimePingMsg(bb, nc, time);
//			if(tl instanceof P2pTimelog) {
//				((P2pTimelog)tl).receivedTimePing(ping.lastTimePongPeriod, ping.sentTime, ping.receivTime);
//			}
			return ping;
		}
		case TimePong:
		{
			TimePongMsg pong = new TimePongMsg(bb, nc); //解包或跳过均可，否则剩下的remain会进入下一轮decode
			//Pblog.IO("p2p:inside pong");
			TimePingMsg msg = pinging.remove(nc);
			if(msg!=null) {
				Timelog tl = nc.getTimelog();
				if(tl instanceof P2pTimelog) {
					((P2pTimelog)tl).receivedTimePong(pong.lastTimePeriod, msg.sentTime, pong.sentTime, tl.getLastReadTime());
				}
			}else {
				throw new IllegalArgumentException("get TimePong but we never ping at "+nc);
			}
			return null;
		}
		case Goodbye:
			return new GoodbyeMsg(bb, nc);
		case PeersWant:
			return new PeersWantMsg(bb, nc);
		case PeersTake:
			return new PeersTakeMsg(bb, nc);
		default:
			throw new IllegalArgumentException("unknow "+funcName()+" code "+(0xFF & b));
		}
	}
	
	/**
	 * 返回true 会暂停读包, 
	 * 可以通过 NormalSession.startRead()恢复读包;
	 * 留意TimeOut,TimeoutRunner依然会自动断开连接的;
	 * 此时可能仍然处于read锁内，但可以使用close()断开链接;
	 * @param Msg msg
	 * @param NormalSession nc
	 * @return true PauseRead;
	 */
	public boolean toDo(Msg msg,NormalSession nc) {
		byte b = ((MsgCodec)msg).getMsgCode();
		switch(b) {
		case TimePing:{
			TimePingMsg ping = (TimePingMsg)msg;
			if(Pblog.DEBUG)Pblog.DEBUG("todo p2p.timePingMsg at ",nc);
			Timelog tl = nc.getTimelog();
			if(tl instanceof P2pTimelog) {
				((P2pTimelog)tl).receivedTimePing(ping.lastTimePongPeriod, ping.sentTime, ping.receivTime);
			}
			nc.writeMsg(new TimePongMsg(ping.receivTime-ping.sentTime));
		}break;
		//case TimePong:
		case Goodbye:{
			GoodbyeMsg goodbye = (GoodbyeMsg)msg;
			if(Pblog.DEBUG)Pblog.DEBUG("todo p2p.goodbyeMsg [",goodbye.func,"]:[",goodbye.reson,"] at ",nc);
			//TODO 加一些关于goodbye的记录
			nc.close();
		}break;
		case PeersWant:{
			PeersWantMsg peerswant = (PeersWantMsg)msg;
			if(Pblog.DEBUG)Pblog.DEBUG("todo p2p.peersWantMsg [key:",Pblog.byteArray2Hex(peerswant.key),"] at ",nc);
			byte[] myKey = peerswant.key.length!=0?peerswant.key:nc.getFunc().getNode().getPubKey();
			Node[] ns = nodeBucket.getNeighbourNodes(myKey, PEERS_TAKE_MAX);
//			if(ns!=null && !ns.isEmpty()) {
//				if(Arrays.equals(myKey, ns.get(0).node.getPubKey())) {
//					ns.remove(0);
//				}
//			}
			PeersTakeMsg peerstake = new PeersTakeMsg(ns);
			nc.writeMsg(peerstake);
		}break;
		case PeersTake:{
			PeersTakeMsg peerstake = (PeersTakeMsg)msg; //返回的nodes里面可能包括自己;
			if(Pblog.DEBUG)Pblog.DEBUG("todo p2p.peersTakeMsg ",peerstake.getNodes().length,":[",peerstake.getNodes().length>0?peerstake.getNodes()[0]:null,",...] at ",nc);
			nodeBucket.addNode(peerstake.getNodes());
		}break;
		default:
			throw new IllegalArgumentException("unknow "+funcName()+" code "+(0xFF & b));
		}
		return false;//msg.getFunc().toDo(msg, nc);
	}
	@Override
	public void sessionTimeout(NormalSession nc) {
		TimePingMsg msg;
		Timelog tl = nc.getTimelog();
		if(tl instanceof P2pTimelog) {
			P2pTimelog ptl = (P2pTimelog)tl;
			msg = new TimePingMsg(ptl.getLastTimePeriod());
		}else if(tl instanceof FuncTimelog) {
			FuncTimelog ptl = (FuncTimelog)tl;
			msg = new TimePingMsg(0-ptl.getRemoteTimeOffset());
		}else {
			msg = new TimePingMsg(0);
		}
		if(null==pinging.putIfAbsent(nc,msg)) {
			nc.writeMsg(msg);
		}
	}


	/**
	 * {@link FuncIndex} cacheindex可以自编解决方案，比如连数据库;
	 * <p>此时不要用close()中断网络(否则会硬解写锁)，throw也仅针对本包, 后续包会继续正常编码;</p>
	 */
	@Override
	public void encode(Msg msg, NormalSession nc, ByteBuffer bb) {
		if(Pblog.DEBUG && msg!=null)Pblog.DEBUG("send ",funcName(),".",msg.getClass().getSimpleName()," at ",nc);
		msg.write(bb, nc);
	}
	
	/**
	 * 255bytes以内;
	 */
	@Override
	public String funcName() {
		return "P2P";
	}
	@Override
	public String toString() {
		//用于出错时找原因;
		return this.getClass().getSimpleName()+"$"+Integer.toHexString(this.hashCode())+"["+this.funcName()+"]";
	}
	
	///////////////////////////////////////////////
	
	private Map<NormalSession,TimePingMsg> pinging;
	public void newSession(NormalSession nc) {
		if(Pblog.DEBUG)Pblog.DEBUG("newSession ",this," in ",nc);
	}
	public void closeSession(NormalSession nc) {
		if(Pblog.DEBUG)Pblog.DEBUG("closeSession ",this," in ",nc);
		pinging.remove(nc);
	}
	public void close(){
		if(Pblog.DEBUG)Pblog.DEBUG("closeFuncProvider ",this);
		this.nodeBucket.removeNewKeyListener(this);
	}

	
	private int index = -1;
	@Override
	public int getIndex() {return index;}
	@Override
	public void setIndex(int index) {this.index = index;}

	/**
	 * 
	 * <p>PeersTakeMsg 返回和本地最接近的一篮子Node;</p>
	 * @author TanYaqiu 
	 * Copyright © 2019 TanYaqiu. All rights reserved.
	 */
	public class PeersTakeMsg extends NodesMsg implements MsgCodec{
		public PeersTakeMsg(ByteBuffer bb, NormalSession nc) {
			init(bb);
		}
		public PeersTakeMsg(Node[] nodes) {
			init(nodes);
		}
		@Override
		public void write(ByteBuffer bb, NormalSession nc) {
			bb.put(getMsgCode());
			super.write(bb, nc);
		}
		
		@Override
		final public FuncProvider getFunc() {
			return P2pFuncProvider.this;
		}
		@Override
		final public byte getMsgCode() {
			return PeersTake;
		}
	}

	
	/**
	 * 报文:
	 * 目标长度[2_bits] 目标[?_bits] 
	 * <p>PeersWantMsg 请求和本地最接近的一篮子Node;</p>
	 * @author TanYaqiu 
	 * Copyright © 2019 TanYaqiu. All rights reserved.
	 */
	public class PeersWantMsg implements MsgCodec{
		byte[] key;
		public PeersWantMsg(ByteBuffer bb, NormalSession nc) {
			key = new byte[0xFFFF & bb.getShort()];
			bb.get(key);
		}
		public PeersWantMsg(byte[] key) {this.key = key;}
		public PeersWantMsg() {}
		@Override
		public void write(ByteBuffer bb, NormalSession nc) {
			bb.put(getMsgCode());
			if(key==null) {
				bb.putShort((short)0);
			}else {
				bb.putShort((short)key.length);
				bb.put(key);
			}
		}
		
		@Override
		final public FuncProvider getFunc() {
			return P2pFuncProvider.this;
		}
		@Override
		final public byte getMsgCode() {
			return PeersWant;
		}
	}

	
	/**
	 * 
	 * <p>GoodbyeMsg 由接收的一方优雅地关闭连接;</p>
	 * <p>* 报文
	 * <br>功能字符长度[1_bit] 功能描叙字符[?_bits]
	 * <br>描叙字符长度[2_bit] 描叙字符[?_bits];</p>
	 * @author TanYaqiu 
	 * Copyright © 2019 TanYaqiu. All rights reserved.
	 */
	public class GoodbyeMsg implements MsgCodec{
		final public String func;
		final public String reson;
		public GoodbyeMsg(ByteBuffer bb, NormalSession nc) {
			this.func = StringUtil.byteBufferToString(1, bb);
			this.reson = StringUtil.byteBufferToString(2, bb);
		}
		public GoodbyeMsg(String func,String reson) {
			this.func = func;
			this.reson= reson;
		}
		@Override
		public void write(ByteBuffer bb, NormalSession nc) {
			//System.err.println("write goodbyeMsg");
			bb.put(getMsgCode());
			StringUtil.stringToByteBuffer(1, func!=null?func:"", bb);
			StringUtil.stringToByteBuffer(2, reson!=null?reson:"", bb);
//			if(first) {
//				while(bb.hasRemaining())bb.putInt(0);
//				bb.putInt(0); //出错测试
//			}else first = true;
		}
		@Override
		final public FuncProvider getFunc() {
			return P2pFuncProvider.this;
		}
		@Override
		final public byte getMsgCode() {
			return Goodbye;
		}
//		@Override
//		public void failed(Throwable exc, NormalSession nc) {
//			if(Pblog.DEBUG)Pblog.DEBUG("GoodbyeMsg canceled in ",nc," cause ",exc);
//		}
//		private boolean first = false;
	}
	
	/**
	 * 
	 * <p>TimePingMsg TimeOut时用来维持链接的，顺便对时</p>
	 * <p>报文: 接收时间[8_bit] 发送时间[8bit]
	 * @author TanYaqiu 
	 * Copyright © 2019 TanYaqiu. All rights reserved.
	 */
	public class TimePingMsg implements MsgCodec{
		long receivTime;
		final long lastTimePongPeriod;
		long sentTime;
		public TimePingMsg(ByteBuffer bb, NormalSession nc, long receivTime) {
			this.receivTime = receivTime;
			this.lastTimePongPeriod = bb.getLong();
			this.sentTime = bb.getLong();	
		}
		public TimePingMsg(long LastTimePongPeriod) {
			this.lastTimePongPeriod = LastTimePongPeriod;
		}
		@Override
		public void write(ByteBuffer bb, NormalSession nc) {
			bb.put(getMsgCode());
			bb.putLong(lastTimePongPeriod);
			bb.putLong(sentTime = System.currentTimeMillis());//发送时才生成,尽量缩短公差;
		}
		
		@Override
		final public FuncProvider getFunc() {
			return P2pFuncProvider.this;
		}
		@Override
		final public byte getMsgCode() {
			return TimePing;
		}
		@Override
		public void failed(Throwable exc, NormalSession nc) {
			pinging.remove(nc);
			if(Pblog.DEBUG)Pblog.DEBUG("TimePing canceled in ",nc," cause ",exc);
		}
	}
	/**
	 * 
	 * <p>TimePongMsg 用于返回TimePingMsg,，</p>
	 * <p>报文: 接收时间[8_bit] 发送时间[8bit]
	 * @author TanYaqiu 
	 * Copyright © 2019 TanYaqiu. All rights reserved.
	 */
	public class TimePongMsg implements MsgCodec{
		final long lastTimePeriod;
		long sentTime;
		public TimePongMsg(ByteBuffer bb, NormalSession nc) {
			lastTimePeriod = bb.getLong();
			sentTime = bb.getLong();	
		}
		public TimePongMsg(long recievTime) {
			this.lastTimePeriod = recievTime;
		}
		@Override
		public void write(ByteBuffer bb, NormalSession nc) {
			bb.put(getMsgCode());
			bb.putLong(lastTimePeriod);
			bb.putLong(System.currentTimeMillis());//发送时才生成,尽量缩短公差;
			//Pblog.IO(Arrays.copyOf(bb.array(), 22));
		}
		
		@Override
		final public FuncProvider getFunc() {
			return P2pFuncProvider.this;
		}
		@Override
		final public byte getMsgCode() {
			return TimePong;
		}
	}

	
	
}
